package com.xiehua.task.actor;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.agent.Agent;
import akka.dispatch.ExecutionContexts;
import com.xiehua.task.actor.cmd.Cmd;
import com.xiehua.task.actor.cmd.NotifyClient;
import com.xiehua.task.job.TaskReverseDefualt;
import com.xiehua.task.job.bucket.DelayBucket;
import com.xiehua.task.job.bucket.TopicBucket;
import com.xiehua.task.job.pool.Job;
import com.xiehua.task.msg.TaskPushReqBody;
import com.xiehua.task.prop.XiehuaConfig;
import com.xiehua.task.protocol.Type;
import com.xiehua.task.protocol.TaskPacket;
import com.xiehua.task.redis.RedisTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.Aio;
import org.tio.core.ChannelContext;
import org.tio.server.ServerGroupContext;
import org.tio.utils.json.Json;
import redis.clients.jedis.Jedis;

import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;

import static com.xiehua.task.actor.cmd.Cmd.CMD_TASK_NOTIFY_ASYNC_RESP;
import static com.xiehua.task.job.pool.Job.JobStatus.notify;


public class TaskNotifyActor extends AbstractActor {

    private static final Logger LOGGER = LoggerFactory.getLogger(TaskNotifyActor.class);

    public Agent<ServerGroupContext> serverGroupContextAgent;

    public TaskNotifyActor(ServerGroupContext serverGroupContext) {
        this.serverGroupContextAgent = Agent.create(serverGroupContext, ExecutionContexts.global());
    }

    @Override
    public AbstractActor.Receive createReceive() {
        return receiveBuilder().match(Cmd.class, t -> t.getCmd().equals(Cmd.CMD_TASK_NOTIFY), s -> {
            if (LOGGER.isDebugEnabled()) LOGGER.debug("收到消息type：{}", s.getCmd());
            Object o = s.getT();
            String jobId = "";
            if (o != null) {
                Job job = (Job) o;
                jobId = job.getJobId();
                notifyClient(job);
            }
            Cmd<String> cmd = new Cmd(CMD_TASK_NOTIFY_ASYNC_RESP, jobId);
            getSender().tell(cmd, getSelf());
        }).matchAny(t -> {
            LOGGER.error("没有对应Actor处理本次消息~");
            unhandled(t);
        }).build();
    }

    private void notifyClient(Job job) {
        Jedis jedis = RedisTemplate.getJedis();
        try {
            //query survival client
            List<NotifyClient> clients = new TopicBucket(jedis).querySurvivalChannelByTopicName(job.getTopic(), serverGroupContextAgent.get());
            if (clients == null || clients.size() < 1) {
                LOGGER.error("没有存活的节点:type->{},taskid->{}", job.getType(), job.getJobId());
                new TaskReverseDefualt (jedis).fail(job.getJobId());//任务调度失败
                ActorSelection emailActor = getContext().actorSelection("/user/EmailActor");
                EmailActor.sendExceptionEmail(new RuntimeException("通知客户端执行任务异常:没有存活的客户端节点,type:"+job.getType()+",taskId:"+job.getJobId()),emailActor, ActorRef.noSender(),job.getTopic());
                return;
            }
            //shuffle list
            Collections.shuffle(clients);
            //change job status
            new TaskReverseDefualt(jedis).notify(job,serverGroupContextAgent.get());
            //build taskPushReqBody
            TaskPushReqBody taskPushReqBody = new TaskPushReqBody();
            taskPushReqBody.setJobId(job.getJobId());
            taskPushReqBody.setType(job.getType());
            taskPushReqBody.setJobContent(Json.toJson(job));
            //build packet
            TaskPacket packet = new TaskPacket(Type.TASK_PUSH_REQ, clients.get(0).getCid());
            String bizData = Json.toJson(taskPushReqBody);
            packet.setBody(bizData.getBytes(TaskPacket.CHARSET));
            packet.signData(bizData, XiehuaConfig.getInstance().get(clients.get(0).getCid()));
            //send notify
            ChannelContext channelContext = Aio.getChannelContextByClientNode(serverGroupContextAgent.get(), clients.get(0).getIp(), clients.get(0).getPort());
            Aio.send(channelContext, packet);
        } catch (Exception e) {
            LOGGER.error("把任务推送到客户端时发生错误:", e);
            ActorSelection emailActor = getContext().actorSelection("/user/EmailActor");
            EmailActor.sendExceptionEmail(e,emailActor, ActorRef.noSender(),job.getTopic());
        } finally {
            RedisTemplate.closeRedis(jedis);
        }


    }


}
